package com.shujia.MR.hw;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * 需求：
 * 将学生总成绩和学生基本信息通过学生ID进行关联
 * <p>
 * 分析：
 * 1.读取数据源，需要读取两份数据
 * 2.根据读取到的数据进行划分 （1500100001	406）  （1500100001,施笑槐,22,女,文科六班）
 * 由于数据中存在不同点，可以根据该特点进行分开处理
 * 3.将数据根据学生ID进行处理，处理结果可以为 Key: 1500100001 Value:406 Key:1500100001 Value:施笑槐,22,女,文科六班
 * 4.由于Reduce端会将相同Key的数据汇总至一起，那么可以实现Join操作
 * 5.将两份数据进行合并，写出
 */

public class JoinMapper extends Mapper<LongWritable, Text, Text, Text> {
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        System.out.println("==>"+key);
        // 通过inputSpit类中的子类FileSplit可以获取到当前文件的路径
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        Path path = inputSplit.getPath();

        boolean flag = path.getName().contains("students.txt");
        String[] split = value.toString().split(",");
        // 长度大于3为学生表
        if (flag) {
            context.write(new Text(split[0]), new Text(split[1] + "\t" + split[2] + "\t" + split[3] + "\t" + split[4]));
        } else {
            context.write(new Text(split[0]), new Text(split[1] + "\t" + split[2]));
        }
    }
}
